package net.oschina.allchat.server.framework.pipe.processing;

import java.io.File;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.lang.reflect.Proxy;
import java.util.ArrayList;
import java.util.Dictionary;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.StringTokenizer;

import net.oschina.allchat.protocol.core.JabberId;
import net.oschina.allchat.protocol.core.Protocol;
import net.oschina.allchat.protocol.core.ProtocolException;
import net.oschina.allchat.protocol.core.stanza.Iq;
import net.oschina.allchat.protocol.core.stanza.Stanza;
import net.oschina.allchat.protocol.core.stanza.error.InternalServerError;
import net.oschina.allchat.protocol.core.stanza.error.ServiceUnavailable;
import net.oschina.allchat.protocol.core.stanza.error.StanzaError;
import net.oschina.allchat.protocol.core.stream.error.BadFormat;
import net.oschina.allchat.protocol.im.stanza.Message;
import net.oschina.allchat.protocol.im.stanza.Presence;
import net.oschina.allchat.server.framework.core.IApplication;
import net.oschina.allchat.server.framework.core.annotations.Component;
import net.oschina.allchat.server.framework.core.annotations.Dependency;
import net.oschina.allchat.server.framework.core.commons.osgi.IBundleContextAware;
import net.oschina.allchat.server.framework.core.commons.osgi.IContributionTracker;
import net.oschina.allchat.server.framework.core.commons.osgi.OsgiUtils;
import net.oschina.allchat.server.framework.core.commons.utils.CommonUtils;
import net.oschina.allchat.server.framework.core.config.DummyConfiguration;
import net.oschina.allchat.server.framework.core.config.IApplicationConfiguration;
import net.oschina.allchat.server.framework.core.config.IApplicationConfigurationAware;
import net.oschina.allchat.server.framework.core.config.IConfiguration;
import net.oschina.allchat.server.framework.core.config.IConfigurationAware;
import net.oschina.allchat.server.framework.core.config.LocalFileConfiguration;
import net.oschina.allchat.server.framework.core.connection.IConnectionContext;
import net.oschina.allchat.server.framework.core.integration.IMessage;
import net.oschina.allchat.server.framework.core.repository.CreationException;
import net.oschina.allchat.server.framework.core.repository.IComponentCollector;
import net.oschina.allchat.server.framework.core.repository.IComponentInfo;
import net.oschina.allchat.server.framework.core.repository.IDependencyManager;
import net.oschina.allchat.server.framework.core.repository.IDependencyManagerAware;
import net.oschina.allchat.server.framework.core.repository.IDestroyable;
import net.oschina.allchat.server.framework.core.repository.IInitializable;

import org.osgi.framework.Bundle;
import org.osgi.framework.BundleContext;
import org.osgi.framework.Constants;
import org.osgi.framework.ServiceRegistration;

@Component("default.protocol.processing.processor")
public class DefaultProtocolProcessingProcessor implements net.oschina.allchat.server.framework.core.integration.IMessageProcessor,
			IConfigurationAware, IBundleContextAware, IContributionTracker, IInitializable, IApplicationConfigurationAware,
				IComponentCollector, IDependencyManager, IDestroyable {
	
	private static final String CONFIGURATION_KEY_STANZA_ERROR_ATTACH_ORIGINAL_MESSAGE = "stanza.error.attach.original.message";
	private static final String STANZA_TYPE_PRESENCE = "presence";
	private static final String STANZA_TYPE_MESSAGE = "message";
	private static final String STANZA_TYPE_IQ = "iq";
	private static final String PROPERTY_NAME_CLASS = "class";
	private static final String PROPERTY_NAME_XEP = "xep";
	private static final String KEY_ALLCHAT_PROCESSORS = "AllChat-Processors";
	private static final String SEPARATOR_PROCESSORS = ",";
	private static final String SEPARATOR_OF_LOCALNAME_NAMESPACE = "|";
	
	protected BundleContext bundleContext;
	
	protected IPresenceProcessor presenceProcessor;
	protected net.oschina.allchat.server.framework.pipe.processing.IMessageProcessor messageProcessor;
	
	protected Map<Xep, Class<IXepProcessor<?, ?>>> processorClasses;
	protected Map<Class<IXepProcessor<?, ?>>, IConfiguration> processorClassAndConfigurations;
	protected Map<Bundle, List<Xep>> bundleAndXeps;
	
	private boolean stanzaErrorAttachOriginalMessage;
	
	private IApplicationConfiguration appConfiguration;
	private String pipeProcessingConfigDir;
	
	private ServiceRegistration<IComponentCollector> srComponentCollector;
	
	private Map<String, IComponentInfo> componentInfos;
	
	public DefaultProtocolProcessingProcessor() {
		processorClasses = new HashMap<Xep, Class<IXepProcessor<?, ?>>>();
		processorClassAndConfigurations = new HashMap<Class<IXepProcessor<?, ?>>, IConfiguration>();
		bundleAndXeps = new HashMap<Bundle, List<Xep>>();
		componentInfos = new HashMap<String, IComponentInfo>();
		
	}

	@Override
	public synchronized void init() {
		Dictionary<String, Object> properties = new Hashtable<String, Object>();
		properties.put(Constants.SERVICE_INTENTS, IApplication.ALLCHAT_FRAMEWORK_COMPONENT_COLLECTOR);
		srComponentCollector = bundleContext.registerService(IComponentCollector.class, this, properties);
		
		OsgiUtils.trackContribution(bundleContext, KEY_ALLCHAT_PROCESSORS, this);
	}
	
	@SuppressWarnings("unchecked")
	@Override
	public void found(Bundle bundle, String contribution) throws Exception {
		StringTokenizer tokenizer = new StringTokenizer(contribution, SEPARATOR_PROCESSORS);
		
		IConfiguration configuration = getConfiguration(bundle);
		List<Xep> xeps = new ArrayList<Xep>();
		while (tokenizer.hasMoreTokens()) {
			String processosrString = tokenizer.nextToken();
			
			Map<String, String> properties = CommonUtils.parsePropertiesString(processosrString,
					new String[] {PROPERTY_NAME_XEP, PROPERTY_NAME_CLASS});
			String sXep = properties.get(PROPERTY_NAME_XEP);
			if (sXep == null) {
				throw new IllegalArgumentException("null xep[register processor]");
			}
			
			Xep xep= parseXep(bundle, sXep);
			
			String sClass = properties.get(PROPERTY_NAME_CLASS);
			if (sClass == null) {
				throw new IllegalArgumentException("null class[register processor]");
			}
			
			Class<?> clazz = bundle.loadClass(sClass);
			
			if (!(IXepProcessor.class.isAssignableFrom(clazz))) {
				throw new RuntimeException(String.format("%s must implement %s[register processor]",
						clazz, IXepProcessor.class));
			}
			
			Class<IXepProcessor<?, ?>> processorClass = (Class<IXepProcessor<?, ?>>)clazz;
			processorClasses.put(xep, processorClass);
			xeps.add(xep);
			processorClassAndConfigurations.put(processorClass, configuration);
		}
		
		bundleAndXeps.put(bundle, xeps);
	}

	private IConfiguration getConfiguration(Bundle bundle) {
		File configurationFile = findConfigurationFile(bundle.getSymbolicName());
		
		if (configurationFile != null) {
			return new LocalFileConfiguration(configurationFile);
		} else {
			return new DummyConfiguration();
		}
	}

	private File findConfigurationFile(String symbolicName) {
		while (true) {
			File file = new File(pipeProcessingConfigDir + "/" + symbolicName + ".ini");
			
			if (file.exists())
				return file;
			
			int dotIndex = symbolicName.indexOf('.');
			
			if (dotIndex == -1)
				return null;
			
			symbolicName = symbolicName.substring(dotIndex + 1);
		}
	}

	private Xep parseXep(Bundle bundle, String sXep) {
		int separatorIndex = sXep.indexOf("->");
		if (separatorIndex == -1 || separatorIndex == sXep.length() - 2) {
			throw new IllegalArgumentException(String.format("invalid xep name %s[register processor]", sXep));
		}
		
		String sStanzaType = sXep.substring(0, separatorIndex);
		Xep.StanzaType stanzaType = getStanzaType(sStanzaType);
		
		if (stanzaType == null)
			throw new IllegalArgumentException(String.format("unknown stanza type %s[register processor]", sStanzaType));
		
		String sProtocol = sXep.substring(separatorIndex + 2);
		Protocol protocol = parseProtocol(sProtocol);
		
		return new Xep(stanzaType, protocol);
	}

	private Xep.StanzaType getStanzaType(String sStanzaType) {
		Xep.StanzaType stanzaType = null;
		
		if (STANZA_TYPE_IQ.equals(sStanzaType)) {
			stanzaType = Xep.StanzaType.IQ;
		} else if (STANZA_TYPE_MESSAGE.equals(sStanzaType)) {
			stanzaType = Xep.StanzaType.MESSAGE;
		} else if (STANZA_TYPE_PRESENCE.equals(sStanzaType)) {
			stanzaType = Xep.StanzaType.PRESENCE;
		}
		
		return stanzaType;
	}

	private Protocol parseProtocol(String sProtocol) {
		String localName;
		String namespace;
		
		int seperator = sProtocol.indexOf(SEPARATOR_OF_LOCALNAME_NAMESPACE);
		if (seperator == -1) {
			throw new IllegalArgumentException(String.format("invalid protocol[%s]", sProtocol));
		}
		
		localName = sProtocol.substring(0, seperator).trim();
		namespace = sProtocol.substring(seperator + 1, sProtocol.length()).trim();
		
		if (localName.length() == 0 || namespace.length() == 0) {
			throw new IllegalArgumentException(String.format("invalid protocol[%s]", sProtocol));
		}
		
		return new Protocol(namespace, localName);
	}

	@Override
	public void lost(Bundle bundle, String contribution) throws Exception {
		List<Xep> xeps = bundleAndXeps.remove(bundle);
		if (xeps != null) {
			for (Xep xep: xeps) {
				Class<IXepProcessor<?, ?>> clazz = processorClasses.remove(xep);
				processorClassAndConfigurations.remove(clazz);
			}
		}
	}

	@Override
	public void setBundleContext(BundleContext bundleContext) {
		this.bundleContext = bundleContext;
	}
	
	@Override
	public void process(IConnectionContext context, IMessage message) {
		try {
			doProcess(context, message);
		} catch (ProtocolException e) {
			context.write(e.getError());
		} catch (RuntimeException e) {
			context.write(new InternalServerError(e.getMessage()));
		}
	}
	
	private class AttachOriginalMessageConnectionContextProxy implements IConnectionContext {
		private IConnectionContext original;
		private String originalMessage;
		
		public AttachOriginalMessageConnectionContextProxy(IConnectionContext original,
				String originalMessage) {
			this.original = original;
			this.originalMessage = originalMessage;
		}

		@Override
		public void setAttribute(Object key, Object value) {
			original.setAttribute(key, value);
		}

		@Override
		public Object getAttribute(Object key) {
			return original.getAttribute(key);
		}

		@Override
		public Object getAttribute(Object key, Object defaultValue) {
			return original.getAttribute(key, defaultValue);
		}

		@Override
		public Object removeAttribute(Object key) {
			return original.removeAttribute(key);
		}

		@Override
		public Object[] getAttributeKeys() {
			return original.getAttributeKeys();
		}

		@Override
		public JabberId getId() {
			return original.getId();
		}

		@Override
		public void write(Object message) {
			if (message instanceof StanzaError) {
				((StanzaError)message).setOriginalMessage(originalMessage);
			}
			
			original.write(message);
		}

		@Override
		public void close() {
			original.close();
		}
		
	}

	private void doProcess(IConnectionContext context, IMessage message) throws RuntimeException {
		Object object = message.getPayload();
		
		if (stanzaErrorAttachOriginalMessage && (object instanceof Stanza)) {
			String originalMessage = ((Stanza)object).getOriginalMessage();
			context = new AttachOriginalMessageConnectionContextProxy(context, originalMessage);
		}
		
		if (object instanceof Presence) {
			Presence presence = (Presence)object;
			if (presence.getObject() == null) {
				processStandardPresence(context, presence);
			} else {
				processXep(context, presence);
			}
		} else if (object instanceof Message) {
			Message msg = (Message)object;
			if (msg.getObject() == null) {
				processStandardMessage(context, msg);
			} else {
				processXep(context, msg);
			}
		} else if (object instanceof Iq) {
			processXep(context, (Iq)object);
		} else {
			context.write(object);
		}
	}

	private void processStandardMessage(IConnectionContext context, Message msg) {
		if (messageProcessor == null || !messageProcessor.process(context, msg)) {
			StanzaError error = new ServiceUnavailable();
			error.setKind(StanzaError.Kind.MESSAGE);
			throw new ProtocolException(error);
		}
	}

	private void processStandardPresence(IConnectionContext context, Presence presence) {
		if (presenceProcessor == null || !presenceProcessor.process(context, presence)) {
			StanzaError error = new ServiceUnavailable();
			error.setKind(StanzaError.Kind.PRESENCE);
			throw new ProtocolException(error);
		}
	}
	
	@SuppressWarnings("unchecked")
	private <K extends Stanza, V> void processXep(IConnectionContext context, Stanza stanza) throws RuntimeException {
		Xep xep= getXep(stanza);
		
		
		Class<IXepProcessor<?, ?>> clazz = processorClasses.get(xep);		
		if (clazz == null) {
			throw new ProtocolException(new BadFormat(String.format("unsupported protocol %s",
					stanza.getObjectProtocol(stanza.getObject()))));
		}
		
		IXepProcessor<K, V> xepProcessor;
		try {
			xepProcessor = (IXepProcessor<K, V>)clazz.newInstance();
		} catch (Exception e) {
			throw new RuntimeException("can't instantiate xep processor", e);
		}
		
		inject(xepProcessor);
		
		if (xepProcessor instanceof IConfigurationAware) {
			((IConfigurationAware)xepProcessor).setConfiguration(processorClassAndConfigurations.get(clazz));
		}
		
		if (xepProcessor instanceof IApplicationConfigurationAware) {
			((IApplicationConfigurationAware)xepProcessor).setApplicationConfiguration(appConfiguration);
		}
		
		if (xepProcessor instanceof IInitializable) {
			((IInitializable)xepProcessor).init();
		}
		
		xepProcessor.process(context, (K)stanza, (V)stanza.getObject());
	}

	private Xep getXep(Stanza stanza) {
		Xep.StanzaType stanzaType = getStanzaType(stanza);
		Protocol protocol = stanza.getObjectProtocol(stanza.getObject());
		
		return new Xep(stanzaType, protocol);
	}

	private Xep.StanzaType getStanzaType(Stanza stanza) {
		if (stanza instanceof Iq)
			return Xep.StanzaType.IQ;
		
		if (stanza instanceof Presence)
			return Xep.StanzaType.PRESENCE;
		
		if (stanza instanceof Message)
			return Xep.StanzaType.MESSAGE;
		
		return null;
	}

	@Dependency("presence.processor")
	public void setPresenceProcessor(IPresenceProcessor presenceProcessor) {
		this.presenceProcessor = presenceProcessor;
		if (this.presenceProcessor instanceof IDependencyManagerAware) {
			((IDependencyManagerAware)this.presenceProcessor).setDependencyManager(this);
		}
	}
	
	@Dependency("message.processor")
	public void setMessageProcessor(net.oschina.allchat.server.framework.pipe.processing.IMessageProcessor messageProcessor) {
		this.messageProcessor = messageProcessor;
		if (this.messageProcessor instanceof IDependencyManagerAware) {
			((IDependencyManagerAware)this.messageProcessor).setDependencyManager(this);
		}
	}
	
	private static class Xep {
		public enum StanzaType {
			IQ,
			PRESENCE,
			MESSAGE
		}
		
		private StanzaType stanzaType;
		private Protocol protocol;
		
		public Xep(StanzaType stanzaType, Protocol protocol) {
			this.stanzaType = stanzaType;
			this.protocol = protocol;
		}
		
		@Override
		public int hashCode() {
			int hash = 7;
			
			hash += 31 * hash + stanzaType.hashCode();
			hash += 31 * hash + protocol.hashCode();
			
			return hash;
		}
		
		@Override
		public boolean equals(Object obj) {
			if (obj == null)
				return false;
			
			if (obj == this)
				return true;
			
			if (obj instanceof Xep) {
				Xep other = (Xep)obj;
				
				if (!(other.stanzaType == this.stanzaType))
					return false;
				
				if (!other.protocol.equals(this.protocol))
					return false;
				
				return true;
			}
			
			return false;
		}
	}

	@Override
	public void setConfiguration(IConfiguration configuration) {
		stanzaErrorAttachOriginalMessage = configuration.getBoolean(
			CONFIGURATION_KEY_STANZA_ERROR_ATTACH_ORIGINAL_MESSAGE,
				false);
	}

	@Override
	public void setApplicationConfiguration(IApplicationConfiguration appConfiguration) {
		this.appConfiguration  = appConfiguration;
		pipeProcessingConfigDir = appConfiguration.getConfigDir() + "/pipe.processing";
	}

	@Override
	public void inject(Object object) {
		Class<?> clazz = object.getClass();
		
		for (Field field : clazz.getDeclaredFields()) {
			Dependency dependencyAnnotation = field.getAnnotation(Dependency.class);
			if (dependencyAnnotation == null) {
				continue;
			}
			
			String id = dependencyAnnotation.value();
			
			ClassLoader oldContextClassLoader = Thread.currentThread().getContextClassLoader();
			Thread.currentThread().setContextClassLoader(clazz.getClassLoader());
			
			Object component = null;
			try {
				component = getComponent(id, field.getType());
			} catch (Exception e) {
				throw new RuntimeException(String.format("can't get dependency component '%s'", id), e);
			} finally {
				Thread.currentThread().setContextClassLoader(oldContextClassLoader);
			}
			
			boolean oldAccessible = field.isAccessible();
			field.setAccessible(true);
			try {
				field.set(object, component);
			} catch (Exception e) {
				throw new RuntimeException(String.format("can't inject dependency '%s'", id), e);
			} finally {
				field.setAccessible(oldAccessible);
			}
			
		}
		
		for (Method method : clazz.getMethods()) {
			if (method.getDeclaringClass().equals(Object.class))
				continue;
			
			int modifiers = method.getModifiers();
			if (!Modifier.isPublic(modifiers) ||
					Modifier.isAbstract(modifiers) ||
					Modifier.isStatic(modifiers))
				continue;
			
			Dependency dependencyAnnotation = method.getAnnotation(Dependency.class);
			if (dependencyAnnotation == null) {
				continue;
			}
			
			String id = dependencyAnnotation.value();
			
			if (isSetterMethod(method)) {
				ClassLoader oldContextClassLoader = Thread.currentThread().getContextClassLoader();
				Thread.currentThread().setContextClassLoader(clazz.getClassLoader());
				
				Object component = null;
				try {
					component = getComponent(id, method.getParameterTypes()[0]);
				} catch (Exception e) {
					throw new RuntimeException(String.format("can't get dependency component '%s'", id), e);
				} finally {
					Thread.currentThread().setContextClassLoader(oldContextClassLoader);
				}
				
				try {
					method.invoke(object, new Object[] {component});
				} catch (Exception e) {
					throw new RuntimeException(String.format("can't inject dependency '%s'", id), e);
				}
			}
		}
		
	}

	private Object getComponent(String id, Class<?> intf) {
		return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[] {intf},
				new LazyLoadComponentInvocationHander(id));
	}
	
	private class LazyLoadComponentInvocationHander implements InvocationHandler {
		private String id;
		private volatile Object component = null;
		
		public LazyLoadComponentInvocationHander(String id) {
			this.id = id;
		}
		
		@Override
		public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
			component = getComponent();
			
			if (component == null)
				throw new IllegalStateException(String.format("null dependency '%s'", id));
			
			return method.invoke(component, args);
		}

		private Object getComponent() throws CreationException {
			if (component != null)
				return component;
			
			synchronized (this) {
				if (component != null)
					return component;
				
				IComponentInfo componentInfo = componentInfos.get(id);
				if (componentInfo == null)
					return null;
				
				return componentInfo.create();
			}
		}
		
	}
		
	private boolean isSetterMethod(Method method) {
		String methodName = method.getName();
		if (methodName.length() > 4 && methodName.startsWith("set") && Character.isUpperCase(methodName.charAt(3))) {
			Class<?>[] types = method.getParameterTypes();
			if (types.length != 1)
				return false;
			
			if (types[0].isPrimitive())
				return false;
			
			return true;
		}
		
		return false;
	}

	@Override
	public synchronized void componentFound(IComponentInfo componentInfo) {
		componentInfos.put(componentInfo.getId(), componentInfo);
	}

	@Override
	public synchronized void componentLost(String componentId) {
		componentInfos.remove(componentId);
	}

	@Override
	public synchronized void destroy() {
		srComponentCollector.unregister();
	}

}
